//===----------------------------------------------------------------------===//
//
//                         BusTub
//
// aggregation_executor.cpp
//
// Identification: src/execution/aggregation_executor.cpp
//
// Copyright (c) 2015-2021, Carnegie Mellon University Database Group
//
//===----------------------------------------------------------------------===//
#include <memory>
#include <utility>
#include <vector>

#include "common/rid.h"
#include "execution/executors/aggregation_executor.h"
#include "storage/table/tuple.h"
#include "type/value.h"

namespace bustub {

AggregationExecutor::AggregationExecutor(ExecutorContext *exec_ctx, const AggregationPlanNode *plan,
                                         std::unique_ptr<AbstractExecutor> &&child)
    : AbstractExecutor(exec_ctx),
      plan_(plan),
      child_(std::move(child)),
      aht_(plan->aggregates_, plan->agg_types_),
      aht_iterator_(aht_.Begin()) {}

void AggregationExecutor::Init() {
  // todo(future): don't clear hash table if executor is called multiple times
  aht_.Clear();
  Tuple tuple{};
  RID dummy_rid{};
  child_->Init();
  while (child_->Next(&tuple, &dummy_rid)) {
    aht_.InsertCombine(MakeAggregateKey(&tuple), MakeAggregateValue(&tuple));
  }
  // aht_iterator_ = aht_.Begin();
  // if(aht_iterator_ == aht_.End()) {
  //   aht_.HandleEmpty();
  // }
  aht_iterator_ = aht_.Begin();
  ending_ = false;
}

auto AggregationExecutor::Next(Tuple *tuple, RID *rid) -> bool {
  if (aht_.Begin() == aht_.End()) {
    if (ending_ || !plan_->group_bys_.empty()) {
      return false;
    }
    *tuple = Tuple{aht_.GenerateInitialAggregateValue().aggregates_, &GetOutputSchema()};
    ending_ = true;
    return true;
  }

  std::vector<Value> values;
  values.reserve(GetOutputSchema().GetColumnCount());
  if (aht_iterator_ == aht_.End()) {
    return false;
  }
  for (auto &key : aht_iterator_.Key().group_bys_) {
    values.push_back(key);
  }
  for (auto &val : aht_iterator_.Val().aggregates_) {
    values.push_back(val);
  }
  *tuple = Tuple{values, &GetOutputSchema()};
  ++aht_iterator_;
  return true;
}

auto AggregationExecutor::GetChildExecutor() const -> const AbstractExecutor * { return child_.get(); }

}  // namespace bustub
